package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

/**
 * @author: xu
 * @desc: 自定义反序列化器
 */
public class MyDeserializationSchemaFunction implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        // 获取变更的数据  value=Struct{after=Struct{id=3,name=ww11,age=55666}
        Struct valueStruct = (Struct) sourceRecord.value();
        Struct sourceStrut = valueStruct.getStruct("source");
        // 获取数据库的名称，表名
        String database = sourceStrut.getString("db");
        String table = sourceStrut.getString("table");

        // 获取操作类型 c->create
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
        if ("create".equals(type)) {
            type = "insert";
        }

        // TODO:获取影响的数据data
        //  源格式：id=1,name=liuxintong,age=17
        //  目标格式：{"id":74603,"order_id":28641,"order_status":"1005","operate_time":"2021-07-30 11:35:49"}}
        Struct afterStruct = valueStruct.getStruct("after");
        // 将变更数据封装为一个json对象
        JSONObject dataJsonObj = new JSONObject();
        if (afterStruct != null) {
            for (Field field : afterStruct.schema().fields()) {
                Object o = afterStruct.get(field);
                dataJsonObj.put(field.name(), o);
            }
        }

        // 创建JSON对象用于封装最终返回值数据信息
        JSONObject result = new JSONObject();
        result.put("database", database);
        result.put("table", table);
        result.put("type", type);
        result.put("data", dataJsonObj);

        // 发送数据至下游
        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
